import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{KeyedStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows

/**
 * 相邻两次数据的 EventTime 的时间差超过指定的时间间隔就会触发执行。如果
 * 加入 Watermark， 会在符合窗口触发的情况下进行延迟。到达延迟水位再进行窗口
 * 触发
 */
object EventTimeSessionWindowsTest {
  def main(args: Array[String]): Unit = {

    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    environment.setParallelism(1)

    val stream = environment.socketTextStream("localhost", 7777)
    val textKeyStream: KeyedStream[(String, Long, Int), Tuple] = stream.map {
      text => {
        val strings = text.split(",")
        (strings(0), strings(1).toLong, 1)
      }
    }.assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.milliseconds(1000)) {
        override def extractTimestamp(element: (String, Long, Int)): Long = {
          element._2
        }
      }
    ).keyBy(_._1)
    textKeyStream.print("textKeyBy:")

    val windowStream = textKeyStream.window(EventTimeSessionWindows.withGap(Time.milliseconds(500)))
    windowStream.reduce(
      (text1, text2) => {
        (text1._1, 0L, text1._3 + text2._3)
      }
    ).map(_._3).print("windows:::").setParallelism(1)


    environment.execute()
  }
}
